package org.paraj.prodcons;

import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.LinkedList;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.TimeUnit;


/**
 * User: piotrga
 * Date: 2007-04-25
 * Time: 20:52:50
 */
public class SimpleEngine<T> implements Engine<T> {

    // Parameters
    protected Collection<Consumer<T>> consumers = new ArrayList<Consumer<T>>();
    protected Collection<Producer<T>> producers = new ArrayList<Producer<T>>();
    private int maxErrorCount = 500;


    // Internal variables
    private Collection<Thread> threads;
    private BlockingQueue<T> queue = new LinkedBlockingQueue<T>();

    // Status flags
    private volatile boolean running = false;
    private volatile boolean stopGracefully = false;

    // Output
    private Collection<ProductError<T>> errors;

    public SimpleEngine<T> addConsumer(Consumer<T> consumer) {
        consumers.add(consumer);
        return this;
    }

    public SimpleEngine<T> addProducer(Producer<T> producer) {
        producers.add(producer);
        return this;
    }

    public synchronized SimpleEngine<T> start() {
        if (running) {
            throw new IllegalStateException("Engine already running");
        }
        errors = Collections.synchronizedCollection(new LinkedList<ProductError<T>>());
        running = true;

        startThreads();
        return this;
    }

    private void startThreads() {
        threads = new ArrayList<Thread>();
        int i = 0;
        for (Consumer<T> c : consumers) threads.add(new Thread(createConsumerThread(c), "Consumer-" + (i++)));
        i = 0;
        for (Producer<T> p : producers) threads.add(new Thread(createProducerThread(p), "Producer-" + (i++)));

        for (Thread t : threads) t.start();
    }

    protected ProducerThread createProducerThread(Producer<T> p) {
        return new ProducerThread(p);
    }

    protected ConsumerThread createConsumerThread(Consumer<T> c) {
        return new ConsumerThread(c);
    }

    public synchronized void stopGracefully() throws InterruptedException {
        stopGracefully = true;
        running = false;
        for (Thread t : threads) t.join();
    }

    public synchronized Collection<T> killEngine() throws InterruptedException {
        if (!running) {
            return queue;
        }

        running = false;
        for (Thread t : threads) t.interrupt();
        for (Thread t : threads) t.join();
        return queue;
    }

    public long pendingSize() {
        return queue.size();
    }

    private void saveError(Throwable t, T product) {
        if (errors.size() >= maxErrorCount) {
//            try{
//                stop();
//            } catch(InterruptedException e){
//                throw new RuntimeException("Error while emergency stopping of the engine", e);
//            }
            throw new EngineException("To many errors. Stopping current thread!");
        }

        errors.add(new ProductError(t, product));
    }

    protected class ProducerThread implements Runnable {
        Producer<T> producer;

        public ProducerThread(Producer<T> producer) {
            this.producer = producer;
        }

        public void run() {
            while (running) {
                try {
                    T product = produce();
                    if(product == null) break;
                    queue.put(product);
                } catch (Throwable t) {
                    saveError(t, null);
                }
            }
        }

        protected T produce() throws Exception {
            return producer.produce();
        }
    }

    protected class ConsumerThread implements Runnable {
        Consumer<T> consumer;

        public ConsumerThread(Consumer<T> consumer) {
            this.consumer = consumer;
        }

        public void run() {
            T product;
            while (running) {
                try {
                    product = queue.poll(100, TimeUnit.MILLISECONDS); // poll is done specifically for stopGracefully method. If it was take() it would wait forever if the queue was empty.
                    if (product != null && running) {
                        consume(product);
                    }
                } catch (Exception e) {
                    saveError(e, null);
                }
            }
            try {
                if (stopGracefully) {
                    while ((product = queue.poll()) != null) {
                        consume(product);
                    }
                }
            } finally {
                closeConsumer();
            }
        }

        protected void closeConsumer() {
            consumer.close();
        }

        protected void consume(T product) {
            try {
                consumer.consume(product);
            } catch (Throwable t) {
                saveError(t, product);
            }
        }
    }

    public Collection<ProductError<T>> getErrors() {
        return errors;
    }

    public void setMaxErrorCount(int maxErrorCount) {
        this.maxErrorCount = maxErrorCount;
    }

}
